/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;

/**
 * @Model cn.com.yang.thread.ParallelFileProcessor
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-下午9:00:03
 */
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

public class ParallelFileProcessor {

    private static final int NUM_THREADS = Runtime.getRuntime().availableProcessors();  // 设置并发线程数，通常根据系统核心数来调整
    private static final String FILE_PATH = "F:\\20gb_data.txt";  // 超大文件路径
    private static final int TOTAL = 5382;  // 每个线程处理的行数 (5000万行 / 8线程)
    private static final int LINES_PER_THREAD = TOTAL / NUM_THREADS;  // 每个线程处理的行数 (5000万行 / 8线程)
    // 全局进度监控
    private static final AtomicLong processedLines = new AtomicLong(0);
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);
        File file = new File(FILE_PATH);
        long fileLength = file.length();
        
        // 创建处理任务
        List<Future<Void>> futures = new ArrayList<>();
        for (int i = 0; i < NUM_THREADS; i++) {
            final long startLine = i * LINES_PER_THREAD;
            final long endLine =  (i == NUM_THREADS - 1) ? TOTAL : (i + 1) * LINES_PER_THREAD-1;
//            if(LINES_PER_THREAD * NUM_THREADS < TOTAL) {
//                futures.add(executorService.submit(new FileProcessorTask(file, LINES_PER_THREAD * NUM_THREADS, TOTAL)));
//            }
            futures.add(executorService.submit(new FileProcessorTask(file, startLine, endLine)));
        }
       
        // 等待所有线程完成
        for (Future<Void> future : futures) {
            future.get();
        }
        
        // 关闭线程池
        executorService.shutdown();
        System.out.println("processedLines:"+processedLines.get());
    }

    static class FileProcessorTask implements Callable<Void> {
        private File file;
        private long startLine;
        private long endLine;

        public FileProcessorTask(File file, long startLine, long endLine) {
            this.file = file;
            this.startLine = startLine;
            this.endLine = endLine;
        }

        @Override
        public Void call() throws Exception {
            try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
                // 跳过文件中的前几个行
                for (long i = 0; i < startLine; i++) {
                    reader.readLine();
                }

                // 处理指定行区间的数据
                for (long i = startLine; i <= endLine; i++) {
                    String line = reader.readLine();
                    if (line != null) {
                        processLine(line);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        }

        private void processLine(String line) {
        	processedLines.incrementAndGet();
            // 处理每一行的逻辑
            System.out.println(Thread.currentThread().getName() + " is processing line: " + line);
        }
    }
}

